/*
 * Copyright (c) 2024, Qi Yaolong.
 * dragon@hbu.edu.cn
 * HeBei University, China.
 * 
 * 简体中文
 * 保留所有权利。除非符合（遵守）以下条款，否则您不得使用此文件：
 * 
 * 禁止任何未经授权的个人或组织以任何形式将本程序及其附属资料、创新、创意、架构设计、算法、衍生作品
 * 等用于任何商业目的、盈利活动、各类竞赛(比赛)、直播教学、录播教学、线下课程、书籍编写、教材编写、
 * 会议、培训、公益活动、项目课题、毕业设计、毕业论文、学术论文等。
 * 
 * 本程序由版权持有人和贡献人“按原样”提供，不提供任何明示或暗示担保，包括但不局限于对适销性和特定
 * 用途适合性的暗示担保。在任何情况下，版权持有人或贡献人对因使用本程序而导致的任何直接、间接、附带、
 * 特殊、典型或后果性损害（包括但不限于购买替代商品或服务；使用损失、数据丢失或利润损失；业务中断）
 * 不承担任何责任，无论其如何产生，也不论其责任理论为何，无论是合同、严格责任还是侵权（包括疏忽或
 * 其他），即使已告知此类损害的可能性。
 * 
 * English
 * All rights reserved. You may NOT use this file except in compliance with the
 * following terms:
 * 
 * IT IS PROHIBITED FOR ANY UNAUTHORIZED PERSON OR ENTITY TO USE THIS PROGRAM AND 
 * ITS ANCILLARY DATA, INNOVATIONS, CONCEPTS, ARCHITECTURE DESIGN, ALGORITHMS, 
 * DERIVED WORKS, ETC., IN ANY FORM FOR COMMERCIAL PURPOSES, PROFIT-MAKING 
 * ACTIVITIES, VARIOUS COMPETITIONS (CONTESTS), LIVE TEACHING, PRE-RECORDED 
 * LESSONS, OFFLINE COURSES, BOOK WRITING, CONFERENCES, TRAININGS, PUBLIC WELFARE 
 * ACTIVITIES, PROJECTS, TOPICS, GRADUATION DESIGNS, DISSERTATIONS, THESES, AND 
 * OTHER APPLICATIONS.
 * 
 * THIS PROGRAM IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS PROGRAM, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

// MQTT接口文件。
// 对paho-mqtt库进行了封装，提供一个简单的接口。
// 使用该接口可以实现MQTT的连接、发布、订阅、断开等功能。

#include <stdio.h>          
#include <stdint.h>         
#include <stdlib.h>         
#include <string.h>         

#include "cmsis_os2.h"      

#include "MQTTClient.h"     
#include "mqtt_ohos.h"      

#include "mqtt_task.h"      

#define LOGI(fmt, ...) printf("[%d] %s " fmt "\n", osKernelGetTickCount(), osThreadGetName(osThreadGetId()), ##__VA_ARGS__)

static volatile int running = 1;

static MQTTClient client = {0};

static Network network = {0};

static unsigned char sendbuf[512], readbuf[512]; 

static void MqttTask(void *arg)
{
    
    LOGI("MqttTask start!");

    MQTTClient *c = (MQTTClient *)arg;

    while (c)
    {
        
        mqttMutexLock(&c->mutex);
        if (!running)
        {
            
            LOGI("MQTT background thread exit!");
            mqttMutexUnlock(&c->mutex);
            break;
        }
        mqttMutexUnlock(&c->mutex);

        mqttMutexLock(&c->mutex);
        if (c->isconnected)
        {
            
            MQTTYield(c, 1);
        }
        mqttMutexUnlock(&c->mutex);

        Sleep(1000);
        
    }
    
    LOGI("MqttTask exit!");
}

void MqttTaskInit(void)
{
    
    NetworkInit(&network);
    MQTTClientInit(&client, &network, 300, sendbuf, sizeof(sendbuf), readbuf, sizeof(readbuf));

    running = 1;

    int rc = ThreadStart(&client.thread, MqttTask, &client);

    LOGI("MqttTaskInit done!");
}

void MqttTaskDeinit(void)
{
    mqttMutexLock(&client.mutex);
    
    running = 0;
    mqttMutexUnlock(&client.mutex);
    mqttMutexDeinit(&client.mutex);
}

int MqttTaskConnect(const char *host, unsigned short port,
                    const char *clientId, const char *username, const char *password)
{
    
    int rc = 0;

    MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;

    if ((rc = NetworkConnect(&network, (char *)host, port)) != 0)
    {
        
        LOGI("NetworkConnect is %d", rc);
        return -1;
    }

    if (username != NULL && password != NULL)
    {
        connectData.username.cstring = (char *)username;
        connectData.password.cstring = (char *)password;
    }

    connectData.MQTTVersion = 3;

    connectData.clientID.cstring = (char *)clientId;

    if ((rc = MQTTConnect(&client, &connectData)) != 0)
    {
        
        LOGI("MQTTConnect failed: %d", rc);
        return -1;
    }

    LOGI("MQTT Connected!");
    return 0;
}

static void OnMessageArrived(MessageData *data)
{
    
    LOGI("Message arrived on topic %.*s: %.*s",
         (int)data->topicName->lenstring.len, (char *)data->topicName->lenstring.data,
         (int)data->message->payloadlen, (char *)data->message->payload);

}

int MqttTaskSubscribe(char *topic)
{
    
    int rc = 0;

    LOGI("Subscribe: [%s] from broker", topic);

    if ((rc = MQTTSubscribe(&client, topic, QOS2, OnMessageArrived)) != 0)
    {
        
        LOGI("MQTTSubscribe failed: %d", rc);
        return -1;
    }

    return 0;
}

int MqttTaskUnSubscribe(char *topic)
{
    
    int rc = 0;

    LOGI("UnSubscribe: [%s] from broker", topic);

    if ((rc = MQTTUnsubscribe(&client, topic)) != 0)
    {
        
        LOGI("MQTTUnsubscribe failed: %d", rc);
        return -1;
    }

    return 0;
}

int MqttTaskPublish(char *topic, char *payload)
{
    
    int rc = 0;

    MQTTMessage message;

    message.qos = QOS2;

    message.retained = 0;

    message.payload = payload;

    message.payloadlen = strlen(payload);

    LOGI("Publish: #'%s': '%s' to broker", topic, payload);

    if ((rc = MQTTPublish(&client, topic, &message)) != 0)
    {
        
        LOGI("MQTTPublish failed: %d", rc);
        return -1;
    }

    return 0;
}

int MqttTaskDisconnect(void)
{
    
    int rc = 0;

    if ((rc = MQTTDisconnect(&client)) != 0)
    {
        
        LOGI("MQTTDisconnect failed: %d", rc);
        return -1;
    }

    NetworkDisconnect(&network);

    return 0;
}
